package org.arneca.hazelcached.impl;

import static java.lang.Integer.parseInt;
import static java.lang.String.valueOf;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Lock;

import com.hazelcast.core.Hazelcast;
import com.thimbleware.jmemcached.LocalCacheElement;
import com.thimbleware.jmemcached.MemCacheDaemon;


public class HazelcastCacheImpl implements com.thimbleware.jmemcached.Cache<LocalCacheElement> {

    String mapName = "default";
    ConcurrentMap<String, LocalCacheElement> map;
    protected AtomicInteger getMisses = new AtomicInteger();
    protected final AtomicLong started = new AtomicLong();

    protected AtomicInteger getCmds = new AtomicInteger();
    protected AtomicInteger setCmds = new AtomicInteger();
    protected AtomicInteger getHits = new AtomicInteger();
    protected AtomicLong casCounter = new AtomicLong(1);
    
    public HazelcastCacheImpl(String _mapName) {
        mapName = _mapName;
        map = Hazelcast.getMap(mapName);
        
    }
    
     static int Now() {
        return (int) (System.currentTimeMillis() / 1000);
    }
    
    
    protected boolean isBlocked(LocalCacheElement e) {
        return e.isBlocked() && e.getBlockedUntil() > Now();
    }

    protected boolean isExpired(LocalCacheElement e) {
        return e.getExpire() != 0 && e.getExpire() < Now();
        
    }
    
    public com.thimbleware.jmemcached.Cache.StoreResponse add(LocalCacheElement _element) {
        try {
            Object  o = map.putIfAbsent(_element.getKeystring(), _element);
            if (o != null)
                return StoreResponse.NOT_STORED;
            return  StoreResponse.STORED;
        } catch (Exception e) {
            return StoreResponse.NOT_STORED;
        }
    }

    public com.thimbleware.jmemcached.Cache.StoreResponse append(LocalCacheElement _element) {
        
        LocalCacheElement old = map.get(_element.getKeystring());
        if (old == null || isBlocked(old) || isExpired(old)) {
            getMisses.incrementAndGet();
            return StoreResponse.NOT_FOUND;
        }
        else {
            int newLength = old.getData().length + _element.getData().length;
            LocalCacheElement replace = new LocalCacheElement(old.getKeystring(), old.getFlags(), old.getExpire());
            ByteBuffer b = ByteBuffer.allocate(newLength);
            b.put(old.getData());
            b.put(_element.getData());
            replace.setData(new byte[newLength]);
            b.flip();
            b.get(replace.getData());
            replace.setCasUnique(replace.getCasUnique() + 1);
            return map.replace(old.getKeystring(), old, replace) ? StoreResponse.STORED : StoreResponse.NOT_STORED;
        }
    }

    public void asyncEventPing() {
        
        
    }

    public com.thimbleware.jmemcached.Cache.StoreResponse cas(Long _cas_key, LocalCacheElement _e) {
        // have to get the element
        LocalCacheElement element = map.get(_e.getKeystring());
        if (element == null || isBlocked(element)) {
            getMisses.incrementAndGet();
            return StoreResponse.NOT_FOUND;
        }

        if (element.getCasUnique() == _cas_key) {
            // casUnique matches, now set the element
            if (map.replace(_e.getKeystring(), element, _e)) return StoreResponse.STORED;
            else {
                getMisses.incrementAndGet();
                return StoreResponse.NOT_FOUND;
            }
        } else {
            // cas didn't match; someone else beat us to it
            return StoreResponse.EXISTS;
        }
    }

    public void close() throws IOException {
        
        
    }

    public com.thimbleware.jmemcached.Cache.DeleteResponse delete(String _key, int _time) {
        
        boolean removed = false;

        // delayed remove
        if (_time != 0) {
           

            // this must go on a queue for processing later...
            LocalCacheElement el = map.get(_key);
            Lock lock = Hazelcast.getLock(el);
            try {
            if (lock.tryLock(_time, TimeUnit.SECONDS)) {
                try {
                    removed = map.remove(el.getKeystring()) != null;
                } finally {
                    lock.unlock();
                }
            } 
            } catch (InterruptedException ie) {
                // Interrupted
            }
            
            
        } else
            removed = map.remove(_key) != null;

        if (removed) return DeleteResponse.DELETED;
        else return DeleteResponse.NOT_FOUND;

       
    }

    public boolean flush_all() {
        map.clear();
        return true;
    }

    public boolean flush_all(int arg0) {
        map.clear();
        return true;
    }

    public LocalCacheElement[] get(String... _keys) {
        getCmds.incrementAndGet();//updates stats

        LocalCacheElement[] elements = new LocalCacheElement[_keys.length];
        int x = 0;
        int hits = 0;
        int misses = 0;
        for (String key : _keys) {
            LocalCacheElement e = map.get(key);
            if (e == null || isExpired(e) || e.isBlocked()) {
                misses++;

                elements[x] = null;
            } else {
                hits++;

                elements[x] = e;
            }
            x++;

        }
        getMisses.addAndGet(misses);
        getHits.addAndGet(hits);

        return elements;
    }

    public long getCurrentBytes() {
        // TODO Should measure the size of the map in bytes
        return 0;
    }

    public long getCurrentItems() {
        return map.size();
    }

    public int getGetCmds() {
        return getCmds.get();
    }

    public int getGetHits() {
           return getHits.get();
    }

    public int getGetMisses() {
       
        return getMisses.get();
    }

    public long getLimitMaxBytes() {
        // TODO Should measure max bytes
        return 0;
    }

    public int getSetCmds() {
        return setCmds.get();
    }

    public Integer get_add(String _key, int _mod) {
        LocalCacheElement old = map.get(_key);
        if (old == null || isBlocked(old) || isExpired(old)) {
            getMisses.incrementAndGet();
            return null;
        } else {
            // TODO handle parse failure!
            int old_val = parseInt(new String(old.getData())) + _mod; // change value
            if (old_val < 0) {
                old_val = 0;

            } // check for underflow

            byte[] newData = valueOf(old_val).getBytes();

            LocalCacheElement replace = new LocalCacheElement(old.getKeystring(), old.getFlags(), old.getExpire());
            replace.setData(newData);
            replace.setCasUnique(replace.getCasUnique() + 1);
            return map.replace(old.getKeystring(), old, replace) ? old_val : null;
        }
    }

    public Set<String> keys() {
        return map.keySet();
    }

    public com.thimbleware.jmemcached.Cache.StoreResponse prepend(LocalCacheElement _element) {
        LocalCacheElement old = map.get(_element.getKeystring());
        if (old == null || isBlocked(old) || isExpired(old)) {
            getMisses.incrementAndGet();
            return StoreResponse.NOT_FOUND;
        }
        else {
            int newLength = old.getData().length + _element.getData().length;

            LocalCacheElement replace = new LocalCacheElement(old.getKeystring(), old.getFlags(), old.getExpire());
            ByteBuffer b = ByteBuffer.allocate(newLength);
            b.put(_element.getData());
            b.put(old.getData());
            replace.setData(new byte[newLength]);
            b.flip();
            b.get(replace.getData());
            replace.setCasUnique(replace.getCasUnique() + 1);
            return map.replace(old.getKeystring(), old, replace) ? StoreResponse.STORED : StoreResponse.NOT_STORED;
        }
    }

    public com.thimbleware.jmemcached.Cache.StoreResponse replace(LocalCacheElement _element) {
        return map.replace(_element.getKeystring(), _element) != null ? StoreResponse.STORED : StoreResponse.NOT_STORED;
    }

    public com.thimbleware.jmemcached.Cache.StoreResponse set(LocalCacheElement _element) {
        setCmds.incrementAndGet();//update stats

        _element.setCasUnique(casCounter.getAndIncrement());

        map.put(_element.getKeystring(), _element);

        return StoreResponse.STORED;
    }

    public Map<String, Set<String>> stat(String _arg) {
      
            Map<String, Set<String>> result = new HashMap<String, Set<String>>();

            if ("keys".equals(_arg)) {
                for (String key : this.keys()) {
                    multiSet(result, "key", key);
                }

                return result;
            }

            // stats we know
            multiSet(result, "version", MemCacheDaemon.memcachedVersion);
            multiSet(result, "cmd_gets", java.lang.String.valueOf(getGetCmds()));
            multiSet(result, "cmd_sets", java.lang.String.valueOf(getSetCmds()));
            multiSet(result, "get_hits", java.lang.String.valueOf(getGetHits()));
            multiSet(result, "get_misses", java.lang.String.valueOf(getGetMisses()));
            multiSet(result, "time", java.lang.String.valueOf(java.lang.String.valueOf(Now())));
            multiSet(result, "uptime", java.lang.String.valueOf(Now() - this.started.longValue()));
            multiSet(result, "cur_items", java.lang.String.valueOf(this.getCurrentItems()));
            multiSet(result, "limit_maxbytes", java.lang.String.valueOf(this.getLimitMaxBytes()));
            multiSet(result, "current_bytes", java.lang.String.valueOf(this.getCurrentBytes()));
            multiSet(result, "free_bytes", java.lang.String.valueOf(Runtime.getRuntime().freeMemory()));

            // Not really the same thing precisely, but meaningful nonetheless. potentially this should be renamed
            multiSet(result, "pid", java.lang.String.valueOf(Thread.currentThread().getId()));

            // stuff we know nothing about; gets faked only because some clients expect this
            multiSet(result, "rusage_user", "0:0");
            multiSet(result, "rusage_system", "0:0");
            multiSet(result, "connection_structures", "0");

            // TODO we could collect these stats
            multiSet(result, "bytes_read", "0");
            multiSet(result, "bytes_written", "0");

            return result;
        
    }
    
    private void multiSet(Map<String, Set<String>> map, String key, String val) {
        Set<String> cur = map.get(key);
        if (cur == null) {
            cur = new HashSet<String>();
        }
        cur.add(val);
        map.put(key, cur);
    }

}
